package com.example.rocketmqconsumer.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Component;

/**
 * @author sherry
 * @description
 * @date Create in 2019/12/23
 * @modified By:
 */
@Component
public class PayConsumer {
    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "pay_consumer_group";
    private static final String TOPIC_NAME = "pay_test_topic";
    private String nameServerAddr = "ROCKET_HOST_NAME_SERVER_ADDR:9876";

    public PayConsumer() {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServerAddr);
//        设置消费地点--从最后一个开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        try {
//            设置订阅的主题，及标签 *，表示订阅所有标签
            consumer.subscribe(TOPIC_NAME, "*");

//            注册监听器
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                try {
                    Message msg = msgs.get(0);
                    System.out.printf("%s 收到消息 %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                    String topic = msg.getTopic();
                    String body = new String(msg.getBody(), "utf-8");
                    String tags = msg.getTags();
                    String keys = msg.getKeys();
                    System.out.println(topic + "\n" + tags + "\n" + keys + "\n" + body);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
